package com.example.es.springbootes.kafka;

import com.alibaba.fastjson.JSON;
import com.example.es.springbootes.entity.Hetong;
import com.example.es.springbootes.utils.EsHlUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import java.util.UUID;

@Component
@Slf4j
public class HetongConsumer {

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private EsHlUtils esHlUtils;

    @Autowired
    private KafkaTemplate kafkaTemplate;

    @KafkaListener(topics = {"productTopic"})
    public void consumer(String message) {
        Hetong hetong = null;
        try {
            hetong = JSON.parseObject(message, Hetong.class);
            //消息重复消费
            if (!redisTemplate.opsForValue().setIfAbsent(hetong.getMsgId(), null)) {
                log.info("该消息已经被消费了：{}",hetong.getId());
                return;
            }
            //保存消息到Es
            esHlUtils.saveOrUpdate(hetong.getId() + "", hetong);
            log.info("ProductConsumer message:{}",message);
        } catch (Exception e) {
            //消息补偿
            kafkaTemplate.send("productTopic", JSON.toJSONString(hetong));
        }
    }
}
